#define DOCTEST_CONFIG_IMPLEMENT_WITH_MAIN
#include "doctest/doctest.h"

#include <functional>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <chrono>
#include "klib/log.hpp"
#include "klib/async.hpp"
#include "klib/ansi_color.hpp"
#include "klib/utils.hpp"

static volatile bool is_end = false;

static const char _HTTP_HEADERS[] = "HTTP/1.1 %u %s\r\nServer: %s\r\n%sContent-Length: %zu\r\n\r\n";

using namespace klib;
using namespace klib::async;
using namespace std::string_view_literals;
using namespace std::string_literals;

TEST_CASE("async") {

    SUBCASE("error_t") {
        struct my_error_t : public error_t {
            using error_t::error_t;
            using error_t::_data;
        };
        
        my_error_t e1;
        CHECK_EQ(nullptr, e1._data);
        CHECK_EQ(0, e1.code());
        
        my_error_t e2("name = {}, age = {}", "kiven", 18);
        CHECK_EQ("name = kiven, age = 18"sv, e2.message());

        e1 = std::move(e2);
        CHECK_EQ("name = kiven, age = 18"sv, e1.message());
        CHECK_EQ(nullptr, e2._data);

        my_error_t e3(std::in_place, "name = %s, age = %d", "kiven", 18);
        CHECK_EQ("name = kiven, age = 18"sv, e3.message());
        CHECK_EQ(1, e3.code());
    }
    
    SUBCASE("result_t") {
        auto r1 = new result_t<void>();
        CHECK_EQ(false, r1->is_completed());
        CHECK_EQ(false, r1->is_succeeded());
        CHECK_EQ(false, r1->is_failed());
        delete r1;

        auto r2 = new result_t<int>(std::make_unique<int>(103));
        CHECK_EQ(true, r2->is_succeeded());
        CHECK_EQ(103, *r2->result());
        auto t = r2->release_result();
        CHECK_EQ(103, *t);
        CHECK_EQ(nullptr, r2->result());
        delete r2;
        
        r2 = new result_t<int>(error_t("name={}", "kiven"));
        CHECK_EQ(true, r2->is_failed());
        CHECK_EQ("name=kiven"sv, r2->cause().message());
        auto e2 = r2->release_cause();
        CHECK_EQ("name=kiven"sv, e2.message());
        delete r2;
    }

    SUBCASE("result_t") {
        auto p1 = new promise_t<void>();
        CHECK_EQ(false, p1->is_completed());
        p1->complete();
        p1->set_handler([] (result_t<void>&& r) {
            CHECK_EQ(true, r.is_succeeded());
        });
        delete p1;

        auto p2 = new promise_t<int>();
        p2->set_handler([] (auto&& r) {
            CHECK_EQ(37, *r.result());
        });
        p2->complete(std::make_unique<int>(37));
        delete p2;
    }

    SUBCASE("future_t") {
        std::condition_variable cv;
        std::mutex mtx;
        bool task_completed = false;
        auto t1 = new future_t<void>();
        CHECK_EQ(false, t1->is_completed());

        t1->promise().complete();

        (*t1)
        .compose([] () {
            future_t<int> t2;
            std::thread([p = t2.promise()] () mutable {
                std::this_thread::sleep_for(std::chrono::milliseconds(1));
                p.complete(125);
            }).detach();
            return t2;
        })
        .compose([] (unique_ptr<int>&& r) {
            CHECK_EQ(125, *r);
            future_t<string> t3;
            std::thread([p = t3.promise()] () mutable {
                std::this_thread::sleep_for(std::chrono::milliseconds(1));
                p.complete("kiven");
            }).detach();
            return t3;
        })
        .compose([] (unique_ptr<string> r) {
            CHECK_EQ("kiven"s, *r);
            future_t<void> t;
            std::thread([p = t.promise()] () mutable {
                std::this_thread::sleep_for(std::chrono::milliseconds(1));
                p.complete();
            }).detach();
            return t;
        })
        .map([] () {
            return unique_ptr<int>(new int(641));
        })
        .map([] (unique_ptr<int> r) {
            CHECK_EQ(641, *r);
        })
        .map([] () {
            return unique_ptr<int>(new int(332));
        })
        .compose([] (unique_ptr<int> r) {
            CHECK_EQ(332, *r);
            return future_success<string>("kiven");
        })
        .compose([] (unique_ptr<string>&& r) {
            CHECK_EQ("kiven"s, *r);
            return future_fail<int>("err={}", 48);
        })
        .otherwise([] (error_t err) {
            CHECK_EQ("err=48"s, err.message());
            future_t<int> t;
            std::thread([p = t.promise()] () mutable {
                std::this_thread::sleep_for(std::chrono::milliseconds(1));
                p.complete(48);
            }).detach();
            return t;
        })
        .on_complete([&mtx, &task_completed, &cv] (result_t<int>&& v) mutable {
            REQUIRE_EQ(true, v.is_succeeded());
            CHECK_EQ(48, *v.result());
            std::unique_lock<std::mutex> lck(mtx);
            task_completed = true;
            cv.notify_all();
        });
        
        delete t1;
        
        // std::this_thread::sleep_for(std::chrono::milliseconds(1000));
        std::unique_lock<std::mutex> lck(mtx);
        while (!task_completed) cv.wait(lck);
    }
    
    SUBCASE("composite_all-success") {
        std::condition_variable cv;
        std::mutex mtx;
        volatile int tc = 0;

        auto t1 = new future_t<void>();
        auto t2 = new future_t<int>();
        auto t3 = new future_t<string>();

        auto th1 = std::thread([p = t1->promise()] () mutable {
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
            p.complete();
        });
        auto th2 = std::thread([p = t2->promise()] () mutable {
            std::this_thread::sleep_for(std::chrono::milliseconds(2));
            p.complete(74);
        });
        auto th3 = std::thread([p = t3->promise()] () mutable {
            std::this_thread::sleep_for(std::chrono::milliseconds(3));
            p.complete("name");
        });
        
        composite_all(*t1, *t2, *t3).on_complete([&mtx, &tc, &cv] (composite_result_t&& res) {
            CHECK_EQ(true, res.is_succeeded());
            auto res_data = res.result();
            CHECK_EQ(true, res_data->at<void>(0).is_succeeded());
            CHECK_EQ(74, *res_data->at<int>(1).result());
            CHECK_EQ("name"s, *res_data->at<string>(2).result());

            std::unique_lock<std::mutex> lck(mtx);
            tc = tc + 1;
            cv.notify_all();
        });

        th1.join();
        th2.join();
        th3.join();
        
        std::unique_lock<std::mutex> lck(mtx);
        while (tc < 1) cv.wait(lck);
        // std::this_thread::sleep_for(std::chrono::microseconds(1000));

        delete t1;
        delete t2;
        delete t3;
    }

    SUBCASE("composite_all-fail") {
        std::condition_variable cv;
        std::mutex mtx;
        volatile int tc = 0;

        auto t1 = new future_t<void>();
        auto t2 = new future_t<int>();
        auto t3 = new future_t<string>();

        auto th1 = std::thread([p = t1->promise()] () mutable {
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
            p.fail("error: {}", 57);
        });
        auto th2 = std::thread([p = t2->promise()] () mutable {
            std::this_thread::sleep_for(std::chrono::milliseconds(2));
            p.complete(74);
        });
        auto th3 = std::thread([p = t3->promise()] () mutable {
            std::this_thread::sleep_for(std::chrono::milliseconds(3));
            p.complete("name");
        });
        
        composite_all(*t1, *t2, *t3).on_complete([&mtx, &tc, &cv] (composite_result_t&& res) {
            CHECK_EQ(true, res.is_failed());
            CHECK_EQ("error: 57"s, res.cause().message());

            std::unique_lock<std::mutex> lck(mtx);
            tc = tc + 1;
            cv.notify_all();
        });

        th1.join();
        th2.join();
        th3.join();
        
        std::unique_lock<std::mutex> lck(mtx);
        while (tc < 1) cv.wait(lck);
        // std::this_thread::sleep_for(std::chrono::microseconds(2000));

        delete t1;
        delete t2;
        delete t3;
    }

    SUBCASE("composite_any-success") {
        std::condition_variable cv;
        std::mutex mtx;
        volatile int tc = 0;

        auto t1 = new future_t<void>();
        auto t2 = new future_t<int>();
        auto t3 = new future_t<string>();

        auto th1 = std::thread([p = t1->promise()] () mutable {
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
            p.fail("error: {}", 57);
        });
        auto th2 = std::thread([p = t2->promise()] () mutable {
            std::this_thread::sleep_for(std::chrono::milliseconds(2));
            p.complete(74);
        });
        auto th3 = std::thread([p = t3->promise()] () mutable {
            std::this_thread::sleep_for(std::chrono::milliseconds(3));
            p.fail("name");
        });
        
        composite_any(*t1, *t2, *t3).on_complete([&mtx, &tc, &cv] (composite_result_t&& res) {
            CHECK_EQ(true, res.is_succeeded());
            CHECK_EQ(74, *res.result()->at<int>(1).result());

            std::unique_lock<std::mutex> lck(mtx);
            tc = tc + 1;
            cv.notify_all();
        });

        delete t1;
        delete t2;
        delete t3;

        th1.join();
        th2.join();
        th3.join();

        std::unique_lock<std::mutex> lck(mtx);
        while (tc < 1) cv.wait(lck);
    }

    SUBCASE("composite_any-fail") {
        std::condition_variable cv;
        std::mutex mtx;
        volatile int tc = 0;

        auto t1 = new future_t<void>();
        auto t2 = new future_t<int>();
        auto t3 = new future_t<string>();

        auto th1 = std::thread([p = t1->promise()] () mutable {
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
            p.fail("error: {}", 57);
        });
        auto th2 = std::thread([p = t2->promise()] () mutable {
            std::this_thread::sleep_for(std::chrono::milliseconds(2));
            p.fail("en");
        });
        auto th3 = std::thread([p = t3->promise()] () mutable {
            std::this_thread::sleep_for(std::chrono::milliseconds(3));
            p.fail("name");
        });
        
        composite_any(*t1, *t2, *t3).on_complete([&mtx, &tc, &cv] (composite_result_t&& res) {
            CHECK_EQ(true, res.is_failed());

            std::unique_lock<std::mutex> lck(mtx);
            tc = tc + 1;
            cv.notify_all();
        });

        delete t1;
        delete t2;
        delete t3;

        th1.join();
        th2.join();
        th3.join();

        std::unique_lock<std::mutex> lck(mtx);
        while (tc < 1) cv.wait(lck);
    }

    SUBCASE("composite_join") {
        std::condition_variable cv;
        std::mutex mtx;
        volatile int tc = 0;

        auto t1 = new future_t<void>();
        auto t2 = new future_t<int>();
        auto t3 = new future_t<string>();

        auto th1 = std::thread([p = t1->promise()] () mutable {
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
            p.fail("error: {}", 57);
        });
        auto th2 = std::thread([p = t2->promise()] () mutable {
            std::this_thread::sleep_for(std::chrono::milliseconds(2));
            p.complete(74);
        });
        auto th3 = std::thread([p = t3->promise()] () mutable {
            std::this_thread::sleep_for(std::chrono::milliseconds(3));
            p.complete("name");
        });
        
        composite_join(*t1, *t2, *t3).on_complete([&mtx, &tc, &cv] (composite_result_t&& res) {
            auto res_data = res.result();
            CHECK_EQ(true, res.is_succeeded());
            CHECK_EQ("error: 57"s, res_data->at<void>(0).cause().message());
            CHECK_EQ(74, *res_data->at<int>(1).result());
            CHECK_EQ("name"s, *res_data->at<string>(2).result());

            std::unique_lock<std::mutex> lck(mtx);
            tc = tc + 1;
            cv.notify_all();
        });

        delete t1;
        delete t2;
        delete t3;

        th1.join();
        th3.join();
        th3.join();

        std::unique_lock<std::mutex> lck(mtx);
        while (tc < 1) cv.wait(lck);
    }

}

